package com.hzw.saas.service.storage.service;

import java.io.File;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;

import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.hzw.saas.common.util.constants.PatternConsts;
import com.hzw.saas.service.storage.model.ResourceFile;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * @author zzl
 * @since 10/22/2021
 */
@Slf4j
public class ResourceMigrateExecutor {

    protected static void execute(Integer threadCount, String taskName, PageFilesHandler pageFilesHandler, DealStorageHandler dealFileHandler) {
        int pageSize = 2000;
        int pageNum = 1;

        // executor
        threadCount = ObjectUtil.compare(threadCount, 0) > 0 ? threadCount : 1;
        ExecutorService executor = ThreadUtil.newExecutor(threadCount);
        Queue<ResourceFile> fileQueue = new ConcurrentLinkedQueue<>();
        AtomicInteger successCount = new AtomicInteger(0);
        AtomicInteger failCount = new AtomicInteger(0);

        // log
        String today = "-" + DateUtil.format(new Date(), PatternConsts.PURE_DATETIME_PATTERN);
        String logSuffix = ".log";
        String logName = taskName + today + logSuffix;
        String errorLogName = taskName + "-error" + today + logSuffix;
        String logRootDir = System.getProperty("user.dir");
        String logDir = "storage-task-log";
        File logFile = FileUtil.file(logRootDir, logDir, logName);
        File errorLogFile = FileUtil.file(logRootDir, logDir, errorLogName);
        FileUtil.touch(logFile);
        FileUtil.touch(errorLogFile);

        String errorTemplate = "catalog: {}, resourceKey: {}, errorMsg: {}";
        List<String> errorLogList = Collections.synchronizedList(new ArrayList<>());

        // task start
        long start = System.currentTimeMillis();
        while (true) {
            IPage<ResourceFile> result = pageFilesHandler.page(new Page<>(pageNum, pageSize));
            if (CollUtil.isEmpty(result.getRecords())) {
                break;
            }
            long stageStart = System.currentTimeMillis();
            long total = result.getTotal();
            long currentPageSize = result.getRecords().size();
            log.info("[{}]文件分段迁移开始, pageNum:{}, pageSize:{}, totalSize: {}, currentPageSize: {}", taskName, pageNum, pageSize, total, currentPageSize);
            fileQueue.addAll(result.getRecords());
            CountDownLatch countDownLatch = new CountDownLatch(threadCount);
            for (int i = 0; i < threadCount; i++) {
                executor.execute(() -> {
                    try {
                        ResourceFile resourceFile;
                        while ((resourceFile = fileQueue.poll()) != null) {
                            try {
                                dealFileHandler.deal(resourceFile);
                                successCount.incrementAndGet();
                                log.debug("[{}]文件迁移成功, catalog: {}, resourceKey: {}", taskName, resourceFile.getCatalog(), resourceFile.getResourceKey());
                            } catch (Exception e) {
                                String errorMsg = StrUtil.format(errorTemplate, resourceFile.getCatalog(), resourceFile.getResourceKey(), e.getMessage());
                                log.error("[{}]文件迁移失败, " + errorMsg, taskName);
                                errorLogList.add(errorMsg);
                                failCount.incrementAndGet();
                            }
                        }
                    } finally {
                        countDownLatch.countDown();
                    }
                });
            }
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                log.error("文件迁移任务被打断");
                throw ExceptionUtil.wrapRuntime(e);
            }
            long stageCost = System.currentTimeMillis() - stageStart;
            String stageLog = StrUtil.format("[{}]文件分段迁移完成, pageNum:{}, pageSize:{}, totalSize: {}, currentTotalSize: {}, failedSize: {}, cost: {}ms", taskName, pageNum, pageSize, total, currentPageSize, errorLogList.size(), stageCost);
            log.info(stageLog);
            FileUtil.appendUtf8Lines(ListUtil.toList(stageLog), logFile);
            if (!errorLogList.isEmpty()) {
                FileUtil.appendUtf8Lines(errorLogList, errorLogFile);
                errorLogList.clear();
            }
            pageNum++;
        }
        long cost = System.currentTimeMillis() - start;
        String endLog = StrUtil.format("[{}]文件迁移任务完成，总数：{}，成功：{}, 失败：{}, 耗时: {}ms", taskName, successCount.get() + failCount.get(), successCount.get(), failCount.get(), cost);
        log.info(endLog);
        FileUtil.appendUtf8Lines(ListUtil.toList(endLog), logFile);
    }

    protected interface PageFilesHandler {
        IPage<ResourceFile> page(Page<?> page);
    }

    protected interface DealStorageHandler {
        void deal(ResourceFile resourceFile);
    }

}
